package org.zhadoop.streaming.config;

import java.util.Locale;
import java.util.Map;

import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.zhadoop.streaming.exception.ErrorCode;
import org.zhadoop.streaming.exception.StreamingException;

import com.google.common.base.Strings;

/**
 * 配置属性中的变量处理
 * @author zwd
 *
 */
public class ConfVariable {
	
	/**
	 * 配置属性中引用变量的前缀
	 */
	public static final String CONF_VARIABLE_PREFIX = "${";
	
	/**
	 * 配置属性中引用变量的后缀
	 */
	public static final String CONF_VARIABLE_POSTFIX = "}";
	
	/**
	 * 配置属性中应用System变量的前缀
	 */
	public static final String SYSTEM_PREFIX = "system:";
	
	/**
	 * 配置属性中引用CQL原有配置属性的前缀
	 */
	public static final String CQLCONF_PREFIX = "conf:";
	
	private static final Logger LOG = LoggerFactory.getLogger(ConfVariable.class);
	
	private ConfValueType type = ConfValueType.COMMON;
	
	private String name;
	
	/**
	 * 分隔符之前的字符串
	 */
	private String preStr = "";
	
	/**
	 * 分隔符之后的字符串
	 */
	private String postStr = "";
	
	/**
	 * 构造函数
	 * 获取配置属性字符串并解析类型和内部名称
	 */
	public ConfVariable(final String value) {
		name = value;
		if(Strings.isNullOrEmpty(value)){
			return;
		}
		String newValue = value.trim();
		int startIndex = newValue.indexOf(CONF_VARIABLE_PREFIX);
		int endIndex = newValue.indexOf(CONF_VARIABLE_POSTFIX);
		
		if(startIndex != -1 && endIndex != -1){
			preStr = newValue.substring(0, startIndex);
			postStr = newValue.substring(endIndex+1);
			
			String variableString = newValue.substring(startIndex,endIndex+1);
			 parseVariableConf(variableString);
		}
		
		//如果以前缀开头，以后缀结尾，才说明是包含变量的配置属性
		//解析过程中发生任何异常，都会还原成普通配置属性
		if(newValue.startsWith(CONF_VARIABLE_PREFIX) && newValue.endsWith(CONF_VARIABLE_POSTFIX)){
			parseVariableConf(newValue);
		}
	}

	
	public static String getValue(ConfVariable confVariable,Map<String,Object> conf,Map<String,String> userConf)
	throws StreamingException{
		String value = null;
		switch(confVariable.type){
			case SYSTEM:
				value = System.getProperty(confVariable.getName());
				break;
			case CONF:
				value = getConfValue(confVariable.getName(),conf,userConf);
				break;
			default:
				return confVariable.getName();
		}
		return confVariable.getPreStr() + value + confVariable.getPostStr();
	}
	
	
	private static String getConfValue(String name, Map<String, Object> conf, Map<String, String> userConf) 
	throws StreamingException{
		if(userConf != null && userConf.containsKey(name)){
			return userConf.get(name);
		}
		if(conf != null && conf.containsKey(name)){
			return conf.get(name).toString();
		}
		StreamingException exception = new StreamingException(ErrorCode.CONFIG_NOT_FOUND,name);
		LOG.error(ErrorCode.CONFIG_NOT_FOUND.getFullMessage(name),exception);
		throw exception;		
	}


	private void parseVariableConf(String value) {
		//直接trim 不会发生空指针，已经测试
		String variable = 
				StringUtils.removeEnd(StringUtils.removeStart(value, CONF_VARIABLE_PREFIX), CONF_VARIABLE_POSTFIX).trim();
		
		if(Strings.isNullOrEmpty(variable)){
			return;
		}
		
		if(variable.toLowerCase(Locale.US).startsWith(SYSTEM_PREFIX)){
			parseSystemVariable(variable);
		}
		
		if(variable.toLowerCase(Locale.US).startsWith(CQLCONF_PREFIX)){
			parseConfVariable(variable);
		}
		
	}



	private void parseConfVariable(String value) {
		String variable = value.substring(CQLCONF_PREFIX.length()).trim();
		
		if(!Strings.isNullOrEmpty(variable)){
			type = ConfValueType.CONF;
			name = variable;
		}
	}


	private void parseSystemVariable(String value) {
		String variable = value.substring(SYSTEM_PREFIX.length()).trim();
		if(!Strings.isNullOrEmpty(variable)){
			type = ConfValueType.SYSTEM;
			name = variable;
		}
	}



	public ConfValueType getType() {
		return type;
	}



	public void setType(ConfValueType type) {
		this.type = type;
	}



	public String getName() {
		return name;
	}



	public void setName(String name) {
		this.name = name;
	}



	public String getPreStr() {
		return preStr;
	}



	public void setPreStr(String preStr) {
		this.preStr = preStr;
	}



	public String getPostStr() {
		return postStr;
	}



	public void setPostStr(String postStr) {
		this.postStr = postStr;
	}
	
	
}
